package bold.module.zk.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.junit.Before;
import org.junit.Test;

public class CuratorLeader {
    CuratorFramework curatorFramework;

    @Before
    public void init() {
        curatorFramework = CuratorFrameworkFactory.builder()
//                .connectString("127.0.0.1:2181")
                .connectString("127.0.0.1:2182")
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .connectionTimeoutMs(15 * 1000) //连接超时时间，默认15秒
                .sessionTimeoutMs(10 * 1000) //会话超时时间，默认60秒   todo 这个时间就是 临时节点的删除时间
//                .namespace("arch") //设置命名空间
                .build();
        curatorFramework.start();
    }

    @Test
    public void leader() throws Exception {

        //CloseMode todo	Latch关闭策略，SILENT-关闭时不触发监听器回调，NOTIFY_LEADER-关闭时触发监听器回调方法，默认不触发
//        LeaderLatch leader = new LeaderLatch(curatorFramework, "/leader", "id", LeaderLatch.CloseMode.SILENT);
        LeaderLatch leader = new LeaderLatch(curatorFramework, "/leader", "id", LeaderLatch.CloseMode.NOTIFY_LEADER);
        LeaderLatchListener leaderLatchListener = new LeaderLatchListener() {

            @Override
            public void notLeader() {
                System.err.println("notLeader");
            }

            @Override
            public void isLeader() {
                System.err.println("isLeader");
            }
        };
        leader.start();
        leader.addListener(leaderLatchListener);


        Thread.currentThread().join();
//        Thread.sleep(10000);
//        leader.close(); 触发 notLeader
    }


    @Test
    public void Leader2() throws Exception {

        LeaderSelectorListener leaderSelectorListener = new LeaderSelectorListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.err.println("stateChanged");
            }

            @Override
            public void takeLeadership(CuratorFramework client) throws Exception {
                System.err.println("takeLeadership");
                Thread.sleep(1000*10);
            }
        };
        LeaderSelector leader = new LeaderSelector(curatorFramework, "/leader", leaderSelectorListener);
        leader.autoRequeue();//可以重复获取  删除当前目录,重新创建节点最小则继续执行
        leader.start();

        Thread.currentThread().join();
//        Thread.sleep(10000);
//        leader.close(); 触发 notLeader
    }
}
